package com.trevor.collector.server;

import com.trevor.collector.CollectorQueue;
import com.trevor.collector.CollectorServer;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Set;

/**
 * <h1></h1>
 * <p>
 * <b>描述：</b>
 * <br>
 * <br>
 * </p>
 *
 * @author xiexiaofeng
 * @version 1.0     2017年09月08日  13点46分
 */
public class SweetCollectorServer implements CollectorServer
{

    private Charset charset = Charset.forName("UTF-8");
    private ServerSocketChannel server = null;
    private Selector selector = null;
    private ByteBuffer buff = ByteBuffer.allocate(1024);
    private int port = 8000;
    private CollectorQueue collectorQueue; // 由Spring注入

    public SweetCollectorServer(int port, CollectorQueue collectorQueue) throws IOException, InterruptedException
    {
        this.port = port;
        this.collectorQueue = collectorQueue;
    }

    public void startServer() throws Exception
    {
        server = ServerSocketChannel.open();
        server.bind(new InetSocketAddress(port));
        server.configureBlocking(false);
        selector = Selector.open();
        server.register(selector, SelectionKey.OP_ACCEPT);
        System.out.println("日志采集服务已启动，监听端口是：" + port);
        listen();
    }


    private void listen() throws Exception
    {
        do
        {
            int wait = selector.select();
            if (wait == 0)
                continue;
            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = keys.iterator();
            while (iterator.hasNext())
            {
                SelectionKey key = iterator.next();
                iterator.remove();
                process(key);
            }
        } while (true);
    }

    /**
     * 容器销毁时调用
     */
    @Override
    public void stop() throws Exception
    {
        if(server != null)
            server.close();
        if (selector != null)
            selector.close();
    }

    public void process(SelectionKey key)
    {
        try
        {
            if (key.isAcceptable())
            {
                ServerSocketChannel server = (ServerSocketChannel) key.channel();
                SocketChannel client = server.accept();
                client.configureBlocking(false);
                client.register(selector, SelectionKey.OP_READ);
                key.interestOps(SelectionKey.OP_ACCEPT);
            }
            if (key.isReadable())
            {
                SocketChannel client = (SocketChannel) key.channel();

                StringBuilder content = new StringBuilder();
                try
                {
                    int count;
                    while ((count = client.read(buff)) > 0)
                    {
                        buff.flip();
                        content.append(charset.decode(buff));
                        buff.compact();
                    }
                    if (count == -1)
                    {
                        System.out.println("断开..."
                                + client.getRemoteAddress());
                        client.close();
                    }
                    key.interestOps(SelectionKey.OP_READ);
                } catch (IOException io)
                {
                    key.cancel();
                    if (key.channel() != null)
                    {
                        key.channel().close();
                    }
                }
                if (content.length() > 0)
                {
                    // 消息入队
                    // System.out.println("收到消息：" + content);
                    collectorQueue.push(client.getRemoteAddress().toString(), content);
                }
            }
        } catch (IOException e)
        {
            System.out.println("客户端关闭连接。。。");
        } catch (Exception e)
        {
            e.printStackTrace();
        }
    }

}
